package com.iamdigger.queldorei.registry.zookeeper;

import com.iamdigger.queldorei.registry.ClosableServiceRegistry;
import com.iamdigger.queldorei.registry.ServiceRegistry;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于 ZooKeeper 的服务注册接口实现
 *
 * @author Sam Tsai
 * @since 0.1.0
 */
public class ZooKeeperServiceRegistry extends ClosableServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceRegistry.class);

    private final ZkClient zkClient;
    private final String zkRegistryPath;

    public ZooKeeperServiceRegistry(String zkAddress, String zkRegistryPath) {
        this(zkAddress, zkRegistryPath, Constant.ZK_CONNECTION_TIMEOUT);
    }

    /**
     * zookeeper连接超时保留可配置.
     *
     * @param zkAddress           zookeeper地址
     * @param zkConnectionTimeout 连接超时时间
     */
    public ZooKeeperServiceRegistry(String zkAddress, String zkRegistryPath, int zkConnectionTimeout) {
        zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, zkConnectionTimeout);
        this.zkRegistryPath = zkRegistryPath;
        LOGGER.debug("connect zookeeper");
    }

    /**
     * 关闭注册服务的zk链接。
     */
    public void closeZkClient() {
        if(null != zkClient) {
            zkClient.close();
        }
    }

    @Override
    public void register(String serviceName, String serviceAddress) {
        // 创建 registry 节点（持久）
        String registryPath = zkRegistryPath;
        if (!zkClient.exists(registryPath)) {
            zkClient.createPersistent(registryPath);
            LOGGER.debug("create registry node: {}", registryPath);
        }
        // 创建 service 节点（持久）
        String servicePath = registryPath + "/" + serviceName;
        if (!zkClient.exists(servicePath)) {
            zkClient.createPersistent(servicePath);
            LOGGER.debug("create service node: {}", servicePath);
        }
        // 创建 address 节点（临时）
        String addressPath = servicePath + "/address-";
        String addressNode = zkClient.createEphemeralSequential(addressPath, serviceAddress);
        LOGGER.debug("create address node: {}", addressNode);
    }

    /**
     * Ephemeral and Sequential nodes will be automatically deleted
     * where client disconnected.
     */
    @Override
    public void afterRegistry() {
        super.afterRegistry();
        //closeZkClient();
    }
}